package cn.ipanel.bigdata.utils

import java.io.InputStream
import java.net.URL
import java.util
import java.lang.reflect.Method
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.dw.Const
import org.apache.spark.sql.catalyst.{CatalystTypeConverters, ScalaReflection}
import org.apache.spark.sql.{DataFrame, Row}

/**
 * Author: lzz
 * Date: 2021/11/17 17:30
 */
case object Util extends Const {

  def getMethod(className: String, methodName: String): Method = Class.forName(className).getMethod(methodName)
  def getClassLoader: ClassLoader = getClass.getClassLoader
  def getResource(sourceName: String): URL = getClassLoader.getResource(sourceName)
  def getResources(sourceName: String): util.Enumeration[URL] = getClassLoader.getResources(sourceName)
  def getResourceAsStream(sourceName: String): InputStream = getClassLoader.getResourceAsStream(sourceName)
  def newInstance(className: String): Any = {
    try {
      Class.forName(className).newInstance()
    } catch {
      case e: Exception => Logger.E(s"new instance failed, because: $e")
    }
  }

  def isEmpty (v: String):    Boolean = if (v == null || "".equals(v)) true else false
  def nonEmpty(v: String):    Boolean = !isEmpty(v)
  def isEmpty (v: Seq[_]): Boolean = if (v == null || v.isEmpty) true else false
  def nonEmpty(v: Seq[_]): Boolean = !isEmpty(v)
  def isEmpty (v: Map[_, _]): Boolean = if (v == null || v.isEmpty) true else false
  def nonEmpty(v: Map[_, _]): Boolean = !isEmpty(v)
  def isEmpty (v: java.util.Map[_, _]): Boolean = if (v == null || v.isEmpty) true else false
  def nonEmpty(v: java.util.Map[_, _]): Boolean = !isEmpty(v)
  def isEmpty (v: java.util.List[_]): Boolean = if (v == null || v.isEmpty) true else false
  def nonEmpty(v: java.util.List[_]): Boolean = !isEmpty(v)
  def isNull(v: String): Boolean = if (isEmpty(v) || "null".equals(v.trim.toLowerCase)) true else false
  def nonNull(v: String): Boolean = !isNull(v)
  def isNull(v: Object): Boolean = if (v == null) true else false
  def nonNull(v: Object): Boolean = !isNull(v)
  def nvlNone(x1: String, x2: String): String = if (isNull(x1) || NONE.equals(x1)) x2 else x1
  def nvlNone(x1: Int, x2: Int): Int = if (INT0 == x1) x2 else x1
  def nvlNone(x1: Long, x2: Long): Long = if (LONG0 == x1) x2 else x1
  def nvlNone(x1: MAP, x2: MAP): MAP = if (isEmpty(x1)) x2 else x1
  def nvlNone(x1: ARR, x2: ARR): ARR = if (isEmpty(x1)) x2 else x1

  private[this] final val NUMBER_PATTEN = """^(-?\d+)$""".r
  private[this] final val DOUBLE_PATTEN = """^(-?\d+)(\.?(\d+)|(\d?))$"""
  private[this] final type ARR = Array[String]
  private[this] final type SEQ = Seq[String]
  private[this] final type MAP = Map[String, String]

  def strToL(v: String, defaultValue: Long = -1L): Long = v match {
    case NUMBER_PATTEN(v) => v.toLong
    case _ => defaultValue
  }
  def strToI(v: String, defaultValue: Int = -1): Int = v match {
    case NUMBER_PATTEN(v) => v.toInt
    case _ => defaultValue
  }

  def strToD(v: String, defaultValue: Double = -1): Double = {
    if (v.matches(DOUBLE_PATTEN)) {
      v.toDouble
    } else{
      defaultValue
    }
  }

  def arrToS(arr: ARR, p: Int, v: String): String = if (nonEmpty(arr) && arr.length > p) nvlNone(arr(p), v) else v
  def arrToI(arr: ARR, p: Int, v: Int): Int = if (nonEmpty(arr) && arr.length > p) strToI(arr(p), v) else v
  def arrToL(arr: ARR, p: Int, v: Long): Long = if (nonEmpty(arr) && arr.length > p) strToL(arr(p), v) else v

  def seqToS(arr: SEQ, p: Int, v: String): String = if (nonEmpty(arr) && arr.length > p) nvlNone(arr(p), v) else v
  def seqToI(arr: SEQ, p: Int, v: Int): Int = if (nonEmpty(arr) && arr.length > p) strToI(arr(p), v) else v
  def seqToL(arr: SEQ, p: Int, v: Long): Long = if (nonEmpty(arr) && arr.length > p) strToL(arr(p), v) else v

  def rowToS(row: Row, k: String, v: String): String = {
    try {
      row.getAs[String](k)
    } catch {
      case _: Exception => v
    }
  }
  def rowToL(row: Row, k: String, v: Long): Long = {
    try {
      row.getAs[Long](k)
    } catch {
      case _: Exception => v
    }
  }
  def rowToI(row: Row, k: String, v: Int): Int = {
    try {
      row.getAs[Int](k)
    } catch {
      case _: Exception => v
    }
  }

  def rowTo[T](row: Row, k: String, v: T): T = {
    try {
      row.getAs[T](k)
    } catch {
      case _: Exception => v
    }
  }

  def mapToS(m: MAP, k: String, v: String): String = if (nonEmpty(m) && m.contains(k)) nvlNone(m(k), v) else v
  def mapToI(m: MAP, k: String, v: Int): Int = if (nonEmpty(m) && m.contains(k)) strToI(m(k), v) else v
  def mapToL(m: MAP, k: String, v: Long): Long = if (nonEmpty(m) && m.contains(k)) strToL(m(k), v) else v
  def mapToD(m: MAP, k: String, v: Double): Double = if (nonEmpty(m) && m.contains(k)) strToD(m(k), v) else v

  //限制取值范围在Array中
  def mapToSLimitRange(m: MAP, k: String, v: String, arr: Array[String]): String = if (nonNull(m) && m.contains(k) && !isNull(m(k)) && arr.contains(m(k))) m(k) else v
  def mapToILimitRange(m: MAP, k: String, v: Int, arr: Array[Int]): Int = if (nonNull(m) && m.contains(k) && arr.contains(strToI(m(k), v))) strToI(m(k), v) else v

  def min(x: Long, y: Long): Long = if (x < y) x else y
  def max(x: Long, y: Long): Long = if (x > y) x else y
  def dif(x: Long, y: Long): Long = max(x, y) - min(x, y)
  def same(x: String, y: String): Boolean = if (nonNull(x)) x.equals(y) else isNull(y)
  def same(x: Int, y: Int): Boolean = x == y
  def same(x: Long, y: Long): Boolean = x == y
  def diff(x: String, y: String): Boolean = !same(x, y)
  def diff(x: Int, y: Int): Boolean = !same(x, y)
  def diff(x: Long, y: Long): Boolean = !same(x, y)

  def transNullAndEmpryAsString(origV: String, defV: String): String = {
    if (origV == null || origV.trim.equals("")) {
      defV
    } else origV
  }

  def intersection(aStart: Long, aEnd: Long, bStart: Long, bEnd: Long): Boolean = {
// if ((aStart >= bStart && aStart < bEnd) || (aStart <= bStart && aEnd > bStart) || (aStart == bStart && aEnd == bEnd)) true
// else false
    if ((aStart < bStart && aEnd > bStart) || (aStart >= bStart && aStart < bEnd)) true
    else false
  }

  object DF {
    import scala.util.{Try, Success, Failure}
    // 判断 dataframe 是否为空 https://blog.csdn.net/BIT_666/article/details/107640484
    def isEmpty (df: DataFrame): Boolean = {
      if (df == null) {
        return true
      }

      Try(df.head()) match {
        case Success(_) => false
        case Failure(_) => true
      }
    }

    def nonEmpty(df: DataFrame): Boolean = !isEmpty(df)

    def dropColumns(df: DataFrame, cols: String*): DataFrame ={
      var newDf = df
      if (nonEmpty(newDf) && cols.nonEmpty) {
        for (e <- cols) {
          newDf = newDf.drop(e)
        }
      }
      newDf
    }

    /**
     * 实现emptyMap，还可以 withColumn("aa", map_fun(array(lit("null"), col("null"))
     * @return
     */
    def empMap(): Map[String, String] = {
      Map[String, String]()
    }

    def map(cols: String*): Map[String, String] = {
      var MAP: Map[String, String] = Map()
      if (cols == null || cols.length % 2 != 0) {
        return MAP
      }

      val keys = cols.indices.filter(_ % 2 == 0).map(cols)
      val values = cols.indices.filter(_ % 2 != 0).map(cols)
      for (i <- 0 until cols.length/2) {
        MAP += (keys(i) -> values(i))
      }

      MAP.filter(x => !x._1.equals("null"))
    }

    def map_concat(mps: Map[String, String]*): Map[String, String] = {
      var MAP: Map[String, String] = Map()
      if (mps == null || mps.isEmpty) {
        return MAP
      }

      for (mp <- mps) {
        if (mp != null && mp.nonEmpty){
          MAP ++= mp
        }
      }

      MAP
    }
  }
}